package com.tdy.cdc.app.dwd;

import com.tdy.cdc.app.BaseSqlApp;
import com.tdy.cdc.common.Constant;
import com.tdy.cdc.util.FlinkSinkUtil;
import com.tdy.cdc.util.FlinkSourceUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamStatementSet;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * Created with IntelliJ IDEA.
 *
 * @Author: NanHuang
 * @Date: 2023/04/03/10:38
 * @Description:
 */
public class App_09_12 extends BaseSqlApp {
    public static void main(String[] args) {
        new App_09_12().init(
                4003,
                "App_09_12",
                2
        );
    }
    @Override
    protected void invoke(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv) {
        // 1 创建源表
        createSourceTables(tableEnv);
        // 2 join
        joinTables(tableEnv);
        // 3 写入Kafka
        writeToKafka(tableEnv);
    }

    private void writeToKafka(StreamTableEnvironment tableEnv) {
        /*tableEnv.executeSql("create table target9 (" +
                "OPER_NURSING_BILL_ID              String,"+
                "HOSPITAL_NAME                     String,"+
                "HOSPITAL_CODE                     String,"+
                "AGE_UNIT                          String,"+
                "AGE                               String,"+
                "SEX_NAME                          String,"+
                "SEX_CODE                          String,"+
                "ID_CARD_NUMBER                    String,"+
                "PATIENT_VISIT_CATEGORY_CODE       String,"+
                "PATIENT_VISIT_CATEGORY            String,"+
                "INPATIENT_VISIT_FLOW_ID           String,"+
                "INPATIENT_NO                      String,"+
                "OUTPATIENT_VISIT_FLOW_ID          String,"+
                "OUTPATIENT_NO                     String,"+
                "PATIENT_ID                        String,"+
                "PATIENT_NAME                      String,"+
                "DOMAIN_ID                         String,"+
                "BED_NAME                          String,"+
                "ROOM_NAME                         String,"+
                "VISIT_TIMES                       String,"+
                "INSTITUTIONAL_INCODE              String,"+
                "INSTITUTIONAL_IN                  String,"+
                "AUTHOR_CODE                       String,"+
                "AUTHOR_NAME                       String,"+
                "CUSTODIAN_CODE                    String,"+
                "CUSTODIAN_NAME                    String,"+
                "PATROL_SIGNDATE                   String,"+
                "VISITATORIAL_NURSE_ID             String,"+
                "VISITATORIAL_NURSE_SIGN_NAME      String,"+
                "INSTRUMENT_NURSE_SIGN_DATETIME    String,"+
                "INSTRUMENT_NURSE_ID               String,"+
                "INSTRUMENT_NURSE_SIGN_NAME        String,"+
                "ASSOCIATE_CHECK_ITEM              String,"+
                "ASSOCIATE_NURSE_ID                String,"+
                "ASSOCIATE_NURSE_SIGN_NAME         String,"+
                "ASSOCIATE_NURSE_SIGN_DATETIME     String,"+
                "TRANSPORT_STAFF_SIGN_DATETIME     String,"+
                "TRANSPORT_STAFF_ID                String,"+
                "TRANSPORT_STAFF_SIGN_NAME         String,"+
                "VISIT_DATE                        String,"+
                "BED_NO                            String,"+
                "ROOM_NO                           String,"+
                "DEPT_ID                           String,"+
                "DEPT_NAME                         String,"+
                "WARD_ID                           String,"+
                "WARD_NAME                         String,"+
                "PATIENT_WEIGHT                    String,"+
                "ABO_BLOOD_CODE                    String,"+
                "ABO_BLOOD_NAME                    String,"+
                "RH_BLOOD_CODE                     String,"+
                "RH_BLOOD_NAME                     String,"+
                "NURSE_GRADE_CODE                  String,"+
                "NURSE_TYPE_CODE                   String,"+
                "NURSE_TYPE_NAME                   String,"+
                "PATROL_DATE_AO                    String,"+
                "PATROL_CODE_AO                    String,"+
                "PATROL_NAME_AO                    String,"+
                "SCRUB_DATE_AO                     String,"+
                "SCRUB_CODE_AO                     String,"+
                "SCRUB_NAME_AO                     String,"+
                "OPER_GOOD_AO                      String,"+
                "PRE_OPER_STOCKTAKE_FLAG           String,"+
                "PATROL_DATE_AC                    String,"+
                "PATROL_CODE_AC                    String,"+
                "SCRUB_CODE_AC                     String,"+
                "SCRUB_NAME_AC                     String,"+
                "IN_OPER_GOODS_NAME                String,"+
                "BEFORE_CHECK_MARK                 String,"+
                "PATROL_DATE_BC                    String,"+
                "PATROL_CODE_BC                    String,"+
                "PATROL_NAME_BC                    String,"+
                "SCRUB_CODE_BC                     String,"+
                "SCRUB_NAME_BC                     String,"+
                "OPER_GOOD_BC                      String,"+
                "AFTER_CHECK_MARK                  String,"+
                "OPER_START_DATETIME               String,"+
                "OPER_END_DATETIME                 String,"+
                "OPER_DOCTOR_ID                    String,"+
                "OPER_DOCTOR_NAME                  String,"+
                "OPER_SITE_NAME                    String,"+
                "OPER_SITE_CODE                    String,"+
                "OPER_CODE                         String,"+
                "OPER_NAME                         String,"+
                "ADMIT_ROOM_DATE                   String,"+
                "OUT_OPER_DATE                     String,"+
                "PREPARATIONS                      String," +
                "XDS_TYPE                          STRING," +
                "XDS_NAME                          STRING," +
                "XDS_VERSION                       STRING," +
                "VISIT_ID                          String,"+
                "table_name                     STRING," +
                "primary key (OPER_NURSING_BILL_ID) not enforced" +
                ")" +
                FlinkSinkUtil.getUpsertKafkaWith(Constant.TOPIC_DWD_MHS_SURGICAL_NURSING_RECORD));*/

        tableEnv.executeSql("create table target10 (" +
                "RECORD_ID                    String,"+
                "HOSPITAL_NAME                String,"+
                "HOSPITAL_CODE                String,"+
                "AGE_UNIT                     String,"+
                "AGE                          String,"+
                "SEX_NAME                     String,"+
                "SEX_CODE                     String,"+
                "ID_CARD_NUMBER               String,"+
                "PATIENT_VISIT_CATEGORY_CODE  String,"+
                "PATIENT_VISIT_CATEGORY       String,"+
                "INPATIENT_VISIT_FLOW_ID      String,"+
                "INPATIENT_NO                 String,"+
                "PATIENT_ID                   String,"+
                "PATIENT_NAME                 String,"+
                "DOMAIN_ID                    String,"+
                "VISIT_DATE                   String,"+
                "BED_NAME                     String,"+
                "ROOM_NAME                    String,"+
                "WARD_ID                      String,"+
                "VISIT_TIMES                  String,"+
                "HEALTH_CARD                  String,"+
                "BED_NO                       String,"+
                "ROOM_NO                      String,"+
                "DEPT_ID                      String,"+
                "DEPT_NAME                    String,"+
                "WARD_NAME                    String,"+
                "AUTHOR_CODE                  String,"+
                "AUTHOR_NAME                  String,"+
                "CUSTODIAN_CODE               String,"+
                "CUSTODIAN_NAME               String,"+
                "DOC_AUDIT_DATE               String,"+
                "DOC_AUDITORCODE              String,"+
                "DOC_AUDITOR                  String,"+
                "DIAG_CODE                    String,"+
                "DIAG_NAME                    String,"+
                "NURSING_GRADECODE            String,"+
                "NURSE_GRADE_NAME             String,"+
                "NURSING_TYPECODE             String,"+
                "NURSE_TYPE_NAME              String,"+
                "NURSING_PROBLEM              String,"+
                "NURSE_NAME                   String,"+
                "NURSE_CATEGORY_ITEM          String,"+
                "NURSE_RESULT                 String,"+
                "CATHETER_DESCRIPTION         String,"+
                "POSTURE_NURSING              String,"+
                "SKIN_CARE                    String,"+
                "TRACHEA_CODE                 String,"+
                "TRACHEA_NAME                 String,"+
                "SAFETY_CODE                  String,"+
                "SAFETY_NAME                  String,"+
                "GUIDE_CODE                   String,"+
                "GUIDE_NAME                   String," +
                "XDS_TYPE                     STRING," +
                "XDS_NAME                     STRING," +
                "XDS_VERSION                  STRING," +
                "VISIT_ID                     String,"+
                "table_name                     STRING," +
                "primary key (RECORD_ID) not enforced" +
                ")"+
                FlinkSinkUtil.getUpsertKafkaWith(Constant.TOPIC_DWD_MHS_CARE_PLAN));

        tableEnv.executeSql("create table  target11(" +
                "RECORD_ID                            String,"+
                "HOSPITAL_NAME                        String,"+
                "HOSPITAL_CODE                        String,"+
                "AGE_UNIT                             String,"+
                "AGE                                  String,"+
                "SEX_NAME                             String,"+
                "SEX_CODE                             String,"+
                "ID_CARD_NUMBER                       String,"+
                "PATIENT_VISIT_CATEGORY_CODE          String,"+
                "PATIENT_VISIT_CATEGORY               String,"+
                "INPATIENT_VISIT_FLOW_ID              String,"+
                "INPATIENT_NO                         String,"+
                "PATIENT_ID                           String,"+
                "PATIENT_NAME                         String,"+
                "DOMAIN_ID                            String,"+
                "VISIT_TIMES                          String,"+
                "CURRENT_DOOR_NUMBER                  String,"+
                "CURRENT_VILLAGE                      String,"+
                "CURRENT_TOWNSHIP                     String,"+
                "CURRENT_COUNTY                       String,"+
                "CURRENT_CITY                         String,"+
                "CURRENT_PROVINCE                     String,"+
                "MARITAL_CODE                         String,"+
                "MARITAL_NAME                         String,"+
                "ETHNIC_CODE                          String,"+
                "ETHNIC                               String,"+
                "OCCUPATION_CODE                      String,"+
                "OCCUPATION                           String,"+
                "DOC_CREATE_TIME                      String,"+
                "DOC_SUBMIT_TIME                      String,"+
                "AUTHOR_CODE                          String,"+
                "AUTHOR_NAME                          String,"+
                "REPRESENTOR_RELATION_CODE            String,"+
                "PRESENTER_ID_NUM                     String,"+
                "RELATIONSHIP_NAME                    String,"+
                "REPRESENTOR_RELIABLE_FLAG            String,"+
                "CASE_REPRESENTOR_NAME                String,"+
                "CUSTODIAN_NAME                       String,"+
                "CUSTODIAN_CODE                       String,"+
                "FINAL_AUDITOR_SIGNDATE               String,"+
                "DOC_FINAL_AUDITORCODE                String,"+
                "DOC_FINAL_AUDITOR                    String,"+
                "ATTEND_SIGNDATE                      String,"+
                "RECEPTION_DOCTOR_ID                  String,"+
                "RECEPTION_DOCTOR_SIGN_NAME           String,"+
                "RESIDENT_SIGNDATE                    String,"+
                "RESIDENT_DOCTOR_ID                   String,"+
                "RESIDENT_DOCTOR_SIGN_NAME            String,"+
                "ATTENDING_SIGNDATE                   String,"+
                "ATTENDING_DOCTOR_ID                  String,"+
                "ATTENDING_DOCTOR_SIGN_NAME           String,"+
                "ADMIT_DATETIME                       String,"+
                "BED_NO                               String,"+
                "BED_NAME                             String,"+
                "ROOM_NO                              String,"+
                "ROOM_NAME                            String,"+
                "DEPT_ID                              String,"+
                "DEPT_NAME                            String,"+
                "WARD_ID                              String,"+
                "WARD_NAME                            String,"+
                "CHIEF_COMPLAINT                      String,"+
                "PRESENT_ILLNESS_HISTORY              String,"+
                "GENERAL_HEALTH_FLAG                  String,"+
                "DISEASES_HISTORY                     String,"+
                "INFECTIOUS_DISEASE_HISTORY           String,"+
                "OBSTETRICAL_HISTORY                  String,"+
                "ALLERGIC_HISTORY                     String,"+
                "OPERATION_HISTORY                    String,"+
                "VACCINATION_HISTORY                  String,"+
                "TRANSFUSION_HISTORY                  String,"+
                "PERSONAL_HISTORY                     String,"+
                "BIRTH_HISTORY                        String,"+
                "MENSTRUAL_HISTORY                    String,"+
                "FAMILY_HISTORY                       String,"+
                "TEMPERATURE_VOLUME                   String,"+
                "PULSE_RATE                           String,"+
                "RESPIRATORY_RATE                     String,"+
                "SYSTOLIC_PRESSURE                    String,"+
                "DIASTOLIC_PRESSURE                   String,"+
                "PATIENT_HIGHT                        String,"+
                "PATIENT_WEIGHT                       String,"+
                "LYMPHONODUS_EXAM_RESULT              String,"+
                "ORGANS_EXAM_RESULT                   String,"+
                "NECK_EXAM_RESULT                     String,"+
                "PE_CHEST_EXAM                        String,"+
                "BELLY_EXAM_RESULT                    String,"+
                "DIGTAL_EXAM_RESULT                   String,"+
                "GENITAL_CHECK_RESULT                 String,"+
                "SPINE_EXAM_RESULT                    String,"+
                "ARMS_EXAM_RESULT                     String,"+
                "NERVOUS_SYSTEM_RESULT                String,"+
                "SPECIALIST_CASE                      String,"+
                "ASSIST_EXAM_TEST                     String,"+
                "CHINA_MED_OBSERVE_RESULT             String,"+
                "THERAPEUTIC_PRINCIPLE                String,"+
                "VISIT_DATE                           String,"+
                "PREL_DIAG_DATE                       String,"+
                "PREL_DIAG_ICD_CODE                   String,"+
                "PREL_DIAG_ICD_NAME                   String,"+
                "DIAG_ORDER                           String,"+
                "PREL_CHINA_DIAG_DATE                 String,"+
                "PREL_CHINA_DISEASE_CODE              String,"+
                "PREL_CHINA_DISEASE_NAME              String,"+
                "DIAG_ORDER_CZY                       String,"+
                "CORRECT_DIAG_DATE                    String,"+
                "CORRECT_DIAG_WESTERN_NAME            String,"+
                "CORRECT_DIAG_WESTERN_CODE            String,"+
                "DIAG_ORDER_XXY                       String,"+
                "CORRECT_CHINA_DIAG_DATE              String,"+
                "CORRECT_DIAG_BMMEDICINE_CODE         String,"+
                "CORRECT_DIAG_BMMEDICINE_NAME         String,"+
                "DIAG_ORDER_XZY                       String,"+
                "SUPPLEMENT_DIAG_DATE                 String,"+
                "SUPPLEMENT_DIAG_WESTERN_NAME         String,"+
                "SUPPLEMENT_DIAG_WESTERN_CODE         String,"+
                "DIAG_ORDER_BXY                       String," +
                "XDS_TYPE                             STRING," +
                "XDS_NAME                             STRING," +
                "XDS_VERSION                          STRING," +
                "VISIT_ID                             String,"+
                "table_name                     STRING," +
                "primary key (RECORD_ID) not enforced" +
                ")"+
                FlinkSinkUtil.getUpsertKafkaWith(Constant.TOPIC_DWD_MHS_IPT_ADMISSIONNOTE));

        tableEnv.executeSql("CREATE TABLE target12 (      "+
                "    RECORD_ID                         STRING    "+
                "    ,HOSPITAL_NAME                    STRING    "+
                "    ,HOSPITAL_CODE                    STRING    "+
                "    ,AGE_UNIT                         STRING    "+
                "    ,AGE                              STRING    "+
                "    ,SEX_NAME                         STRING    "+
                "    ,SEX_CODE                         STRING    "+
                "    ,ID_CARD_NUMBER                   STRING    "+
                "    ,PATIENT_VISIT_CATEGORY_CODE      STRING    "+
                "    ,PATIENT_VISIT_CATEGORY           STRING    "+
                "    ,INPATIENT_VISIT_FLOW_ID          STRING    "+
                "    ,INPATIENT_NO                     STRING    "+
                "    ,PATIENT_ID                       STRING    "+
                "    ,PATIENT_NAME                     STRING    "+
                "    ,DOMAIN_ID                        STRING    "+
                "    ,VISIT_TIMES                      STRING    "+
                "    ,BIRTH_DATE                       STRING    "+
                "    ,AUTHOR_CODE                      STRING    "+
                "    ,AUTHOR_NAME                      STRING    "+
                "    ,CUSTODIAN_CODE                   STRING    "+
                "    ,CUSTODIAN_NAME                   STRING    "+
                "    ,SENIOR_DOCTORE_SIGN_TIME         STRING    "+
                "    ,SENIOR_DOCTOR_ID                 STRING    "+
                "    ,SENIOR_DOCTOR_SIGN_NAME          STRING    "+
                "    ,RESIDENT_DOCTORE_SIGN_TIME       STRING    "+
                "    ,RESIDENT_DOCTOR_ID               STRING    "+
                "    ,RESIDENT_DOCTOR_SIGN_NAME        STRING    "+
                "    ,VISIT_DATE                       STRING    "+
                "    ,BED_NO                           STRING    "+
                "    ,BED_NAME                         STRING    "+
                "    ,ROOM_NO                          STRING    "+
                "    ,ROOM_NAME                        STRING    "+
                "    ,WARD_ID                          STRING    "+
                "    ,WARD_NAME                        STRING    "+
                "    ,DEPT_ID                          STRING    "+
                "    ,DEPT_NAME                        STRING    "+
                "    ,CHIEF_COMPLAINT                  STRING    "+
                "    ,CASE_CHARACTER                   STRING    "+
                "    ,CHINA_MED_OBSERVE_RESULT         STRING    "+
                "    ,DIAG_BASIS                       STRING    "+
                "    ,TREAT_PLAN                       STRING    "+
                "    ,THERAPEUTIC_PRINCIPLE            STRING    "+
                "    ,PREL_DIAG_ICD_CODE               STRING    "+
                "    ,PREL_DIAG_ICD_NAME               STRING    "+
                "    ,PREL_CHINA_DISEASE_CODE          STRING    "+
                "    ,PREL_CHINA_DISEASE_NAME          STRING    "+
                "    ,PREL_CHINA_SYNDROME_CODE         STRING    "+
                "    ,PREL_CHINA_SYNDROME_NAME         STRING    "+
                "    ,DIFF_DIAG_ICD_CODE               STRING    "+
                "    ,DIFF_DIAG_ICD_NAME               STRING    "+
                "    ,DIFF_CHINA_DISEASE_CODE          STRING    "+
                "    ,DIFF_CHINA_DISEASE_NAME          STRING    "+
                "    ,DIFF_CHINA_SYNDROME_CODE         STRING    "+
                "    ,DIFF_CHINA_SYNDROME_NAME         STRING    " +
                "    ,XDS_TYPE                         STRING    " +
                "    ,XDS_NAME                         STRING    " +
                "    ,XDS_VERSION                      STRING    " +
                "    ,VISIT_ID                         STRING    "+
                "    ,table_name                       STRING    " +
                "    ,PRIMARY KEY (RECORD_ID) NOT ENFORCED       "+
                ") "+ FlinkSinkUtil.getUpsertKafkaWith(Constant.TOPIC_DWD_MHS_IPT_FIRST_COD_RECORD));

        // 导入数据
        StreamStatementSet statementSet = tableEnv.createStatementSet();
        statementSet
//                .addInsertSql("insert into target9 select * from join_result9")
                .addInsertSql("insert into target10 select * from join_result10")
                .addInsertSql("insert into target11 select * from join_result11")
                .addInsertSql("insert into target12 select * from join_result12");
        statementSet.execute();
    }

    private void joinTables(StreamTableEnvironment tableEnv) {
        //1 设置数据的TTL（暂时10min）
        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(30 * 60));
        //2 join操作
/*        Table joinResult9 = tableEnv.sqlQuery("select " +
                "s.unique_id                                                      AS OPER_NURSING_BILL_ID,"+
                "hospital_name                                                    AS HOSPITAL_NAME,"+
                "hospital_code                                                    AS HOSPITAL_CODE,"+
                "age_unit                                                         AS AGE_UNIT,"+
                "age                                                              AS AGE,"+
                "sex                                                              AS SEX_NAME,"+
                "sex_code                                                         AS SEX_CODE,"+
                "identify_no                                                      AS ID_CARD_NUMBER,"+
                "patient_typecode                                                 AS PATIENT_VISIT_CATEGORY_CODE,"+
                "patient_type                                                     AS PATIENT_VISIT_CATEGORY,"+
                "case when patient_typecode ='3' then visit_id else '' end       AS INPATIENT_VISIT_FLOW_ID,"+
                "inpatient                                                        AS INPATIENT_NO,"+
                "case when patient_typecode <> '3' then visit_id else '' end     AS OUTPATIENT_VISIT_FLOW_ID,"+
                "out_patient_id                                                   AS OUTPATIENT_NO,"+
                "patient_id                                                       AS PATIENT_ID,"+
                "patient_name                                                     AS PATIENT_NAME,"+
                "s.domain_id                                                      AS DOMAIN_ID,"+
                "bed_name                                                         AS BED_NAME,"+
                "ward_name                                                        AS ROOM_NAME,"+
                "visit_times                                                      AS VISIT_TIMES,"+
                "institutional_incode                                             AS INSTITUTIONAL_INCODE,"+
                "institutional_in                                                 AS INSTITUTIONAL_IN,"+
                "author_code                                                      AS AUTHOR_CODE,"+
                "author_name                                                      AS AUTHOR_NAME,"+
                "custodian_code                                                   AS CUSTODIAN_CODE,"+
                "custodian_name                                                   AS CUSTODIAN_NAME,"+
                "patrol_signdate                                                  AS PATROL_SIGNDATE,"+
                "patrol_code                                                      AS VISITATORIAL_NURSE_ID,"+
                "patrol_name                                                      AS VISITATORIAL_NURSE_SIGN_NAME,"+
                "instrument_signdate                                              AS INSTRUMENT_NURSE_SIGN_DATETIME,"+
                "instrument_code                                                  AS INSTRUMENT_NURSE_ID,"+
                "instrument_name                                                  AS INSTRUMENT_NURSE_SIGN_NAME,"+
                "handover_checkup                                                 AS ASSOCIATE_CHECK_ITEM,"+
                "handover_code                                                    AS ASSOCIATE_NURSE_ID,"+
                "handover_name                                                    AS ASSOCIATE_NURSE_SIGN_NAME,"+
                "handover_signdate                                                AS ASSOCIATE_NURSE_SIGN_DATETIME,"+
                "s.shipper_signdate                                                 AS TRANSPORT_STAFF_SIGN_DATETIME,"+
                "s.shipper_code                                                     AS TRANSPORT_STAFF_ID,"+
                "s.shipper_name                                                     AS TRANSPORT_STAFF_SIGN_NAME,"+
                "visit_date                                                       AS VISIT_DATE,"+
                "bed_no                                                           AS BED_NO,"+
                "ward_id                                                          AS ROOM_NO,"+
                "dept_code                                                        AS DEPT_ID,"+
                "dept_name                                                        AS DEPT_NAME,"+
                "wards_id                                                         AS WARD_ID,"+
                "wards_name                                                       AS WARD_NAME,"+
                "weight                                                           AS PATIENT_WEIGHT,"+
                "abo_code                                                         AS ABO_BLOOD_CODE,"+
                "abo                                                              AS ABO_BLOOD_NAME,"+
                "rh_code                                                          AS RH_BLOOD_CODE,"+
                "rh                                                               AS RH_BLOOD_NAME,"+
                "nursing_gradecode                                                AS NURSE_GRADE_CODE,"+
                "nursing_typecode                                                 AS NURSE_TYPE_CODE,"+
                "nursing_type                                                     AS NURSE_TYPE_NAME,"+
                "patrol_date_ao                                                   AS PATROL_DATE_AO,"+
                "patrol_code_ao                                                   AS PATROL_CODE_AO,"+
                "patrol_name_ao                                                   AS PATROL_NAME_AO,"+
                "scrub_date_ao                                                    AS SCRUB_DATE_AO,"+
                "scrub_code_ao                                                    AS SCRUB_CODE_AO,"+
                "scrub_name_ao                                                    AS SCRUB_NAME_AO,"+
                "oper_good_ao                                                     AS OPER_GOOD_AO,"+
                "preoperative_counting_mark                                       AS PRE_OPER_STOCKTAKE_FLAG,"+
                "patrol_date_ac                                                   AS PATROL_DATE_AC,"+
                "patrol_code_ac                                                   AS PATROL_CODE_AC,"+
                "scrub_code_ac                                                    AS SCRUB_CODE_AC,"+
                "scrub_name_ac                                                    AS SCRUB_NAME_AC,"+
                "oper_good_ac                                                     AS IN_OPER_GOODS_NAME,"+
                "before_check_mark                                                AS BEFORE_CHECK_MARK,"+
                "patrol_date_bc                                                   AS PATROL_DATE_BC,"+
                "patrol_code_bc                                                   AS PATROL_CODE_BC,"+
                "patrol_name_bc                                                   AS PATROL_NAME_BC,"+
                "scrub_code_bc                                                    AS SCRUB_CODE_BC,"+
                "scrub_name_bc                                                    AS SCRUB_NAME_BC,"+
                "oper_good_bc                                                     AS OPER_GOOD_BC,"+
                "after_check_mark                                                 AS AFTER_CHECK_MARK,"+
                "oper_start_date                                                  AS OPER_START_DATETIME,"+
                "oper_end_date                                                    AS OPER_END_DATETIME,"+
                "surgeon_code                                                     AS OPER_DOCTOR_ID,"+
                "surgeon_name                                                     AS OPER_DOCTOR_NAME,"+
                "target_code                                                      AS OPER_SITE_CODE,"+
                "target_name                                                      AS OPER_SITE_NAME,"+
                "oper_code                                                        AS OPER_CODE,"+
                "oper_name                                                        AS OPER_NAME,"+
                "admit_room_date                                                  AS ADMIT_ROOM_DATE,"+
                "out_oper_date                                                    AS OUT_OPER_DATE,"+
                "preparations                                                     AS PREPARATIONS," +
                "s.xds_type as XDS_TYPE," +
                "s.xds_name as XDS_NAME," +
                "s.xds_version as XDS_VERSION," +
                "visit_id                                                         AS VISIT_ID," +
                "'dwd_mhs_surgical_nursing_record' as table_name "+
                "from hdsd00_08_05 s " +
                "left join tdy_list_oper b on b.unique_id=s.unique_id and b.xds_type=s.xds_type " +
                "left join tdy_list_handover c on c.unique_id=s.unique_id and c.xds_type=s.xds_type");
        tableEnv.createTemporaryView("join_result9",joinResult9);*/

        Table joinResult10 = tableEnv.sqlQuery("select " +
                "s.unique_id             AS RECORD_ID," +
                "hospital_name         AS HOSPITAL_NAME," +
                "hospital_code         AS HOSPITAL_CODE," +
                "age_unit              AS AGE_UNIT," +
                "age                   AS AGE," +
                "sex                   AS SEX_NAME," +
                "sex_code              AS SEX_CODE," +
                "identify_no           AS ID_CARD_NUMBER," +
                "patient_typecode      AS PATIENT_VISIT_CATEGORY_CODE," +
                "patient_type          AS PATIENT_VISIT_CATEGORY," +
                "visit_id              AS INPATIENT_VISIT_FLOW_ID," +
                "inpatient             AS INPATIENT_NO," +
                "patient_id            AS PATIENT_ID," +
                "patient_name          AS PATIENT_NAME," +
                "s.domain_id             AS DOMAIN_ID," +
                "visit_date            AS VISIT_DATE," +
                "bed_name              AS BED_NAME," +
                "ward_name             AS ROOM_NAME," +
                "wards_id              AS WARD_ID," +
                "visit_times           AS VISIT_TIMES," +
                "health_card           AS HEALTH_CARD," +
                "bed_no                AS BED_NO," +
                "ward_id               AS ROOM_NO," +
                "dept_code             AS DEPT_ID," +
                "dept_name             AS DEPT_NAME," +
                "wards_name            AS WARD_NAME," +
                "author_code           AS AUTHOR_CODE," +
                "author_name           AS AUTHOR_NAME," +
                "custodian_code        AS CUSTODIAN_CODE," +
                "custodian_name        AS CUSTODIAN_NAME," +
                "doc_audit_date        AS DOC_AUDIT_DATE," +
                "doc_auditorcode       AS DOC_AUDITORCODE," +
                "doc_auditor           AS DOC_AUDITOR," +
                "diag_code             AS DIAG_CODE," +
                "diag_name             AS DIAG_NAME," +
                "nursing_gradecode     AS NURSING_GRADECODE," +
                "nursing_grade         AS NURSE_GRADE_NAME," +
                "nursing_typecode      AS NURSING_TYPECODE," +
                "nursing_type          AS NURSE_TYPE_NAME," +
                "nursing_problem       AS NURSING_PROBLEM," +
                "nurse_name            AS NURSE_NAME," +
                "nurse_category_item   AS NURSE_CATEGORY_ITEM," +
                "nurse_result          AS NURSE_RESULT," +
                "catheter_description  AS CATHETER_DESCRIPTION," +
                "posture_nursing       AS POSTURE_NURSING," +
                "skin_care             AS SKIN_CARE," +
                "c.code                AS TRACHEA_CODE," +
                "c.name                AS TRACHEA_NAME," +
                "b.code                AS SAFETY_CODE," +
                "b.name                AS SAFETY_NAME," +
                "a.code                AS GUIDE_CODE," +
                "a.name                AS GUIDE_NAME," +
                "s.xds_type as XDS_TYPE," +
                "s.xds_name as XDS_NAME," +
                "s.xds_version as XDS_VERSION," +
                "visit_id              AS VISIT_ID," +
                "'dwd_mhs_care_plan' as table_name " +
                "from hdsd00_09_03 s " +
                "left join tdy_list_diet_guide a  on a.unique_id=s.unique_id and a.xds_type=s.xds_type  " +
                "left join tdy_list_nurse_safety b  on b.unique_id=s.unique_id and b.xds_type=s.xds_type  " +
                "left join tdy_list_nurse_trachea c on c.unique_id=s.unique_id and c.xds_type=s.xds_type  " +
                "");
        tableEnv.createTemporaryView("join_result10",joinResult10);

        Table joinResult11 = tableEnv.sqlQuery("select " +
                "s.unique_id                                                          AS RECORD_ID,"+
                "hospital_name                                                        AS HOSPITAL_NAME,"+
                "hospital_code                                                        AS HOSPITAL_CODE,"+
                "age_unit                                                             AS AGE_UNIT,"+
                "age                                                                  AS AGE,"+
                "sex                                                                  AS SEX_NAME,"+
                "sex_code                                                             AS SEX_CODE,"+
                "identify_no                                                          AS ID_CARD_NUMBER,"+
                "patient_typecode                                                     AS PATIENT_VISIT_CATEGORY_CODE,"+
                "patient_type                                                         AS PATIENT_VISIT_CATEGORY,"+
                "visit_id                                                             AS INPATIENT_VISIT_FLOW_ID,"+
                "inpatient                                                            AS INPATIENT_NO,"+
                "patient_id                                                           AS PATIENT_ID,"+
                "patient_name                                                         AS PATIENT_NAME,"+
                "s.domain_id                                                          AS DOMAIN_ID,"+
                "visit_times                                                          AS VISIT_TIMES,"+
                "current_door_number                                                  AS CURRENT_DOOR_NUMBER,"+
                "current_village                                                      AS CURRENT_VILLAGE,"+
                "current_township                                                     AS CURRENT_TOWNSHIP,"+
                "current_county                                                       AS CURRENT_COUNTY,"+
                "current_city                                                         AS CURRENT_CITY,"+
                "current_province                                                     AS CURRENT_PROVINCE,"+
                "marital_code                                                         AS MARITAL_CODE,"+
                "marital_name                                                         AS MARITAL_NAME,"+
                "ethnic_code                                                          AS ETHNIC_CODE,"+
                "ethnic                                                               AS ETHNIC,"+
                "occupation_code                                                      AS OCCUPATION_CODE,"+
                "occupation                                                           AS OCCUPATION,"+
                "write_time                                                           AS DOC_CREATE_TIME,"+
                "effective_time                                                       AS DOC_SUBMIT_TIME,"+
                "author_code                                                          AS AUTHOR_CODE,"+
                "author_name                                                          AS AUTHOR_NAME,"+
                "relationship_code                                                    AS REPRESENTOR_RELATION_CODE,"+
                "presenter_id_num                                                     AS PRESENTER_ID_NUM,"+
                "relationship_name                                                    AS RELATIONSHIP_NAME,"+
                "statement_reliability_mark                                           AS REPRESENTOR_RELIABLE_FLAG,"+
                "history_narrator                                                     AS CASE_REPRESENTOR_NAME,"+
                "custodian_code                                                       AS CUSTODIAN_NAME,"+
                "custodian_name                                                       AS CUSTODIAN_CODE,"+
                "final_auditor_signdate                                               AS FINAL_AUDITOR_SIGNDATE,"+
                "doc_final_auditorcode                                                AS DOC_FINAL_AUDITORCODE,"+
                "doc_final_auditor                                                    AS DOC_FINAL_AUDITOR,"+
                "attend_signdate                                                      AS ATTEND_SIGNDATE,"+
                "s.admission_code                                                     AS RECEPTION_DOCTOR_ID,"+
                "s.admission_name                                                     AS RECEPTION_DOCTOR_SIGN_NAME,"+
                "resident_signdate                                                    AS RESIDENT_SIGNDATE,"+
                "resident_code                                                        AS RESIDENT_DOCTOR_ID,"+
                "resident_name                                                        AS RESIDENT_DOCTOR_SIGN_NAME,"+
                "attending_signdate                                                   AS ATTENDING_SIGNDATE,"+
                "attending_code                                                       AS ATTENDING_DOCTOR_ID,"+
                "attending_name                                                       AS ATTENDING_DOCTOR_SIGN_NAME,"+
                "admission_date                                                       AS ADMIT_DATETIME,"+
                "bed_no                                                               AS BED_NO,"+
                "bed_name                                                             AS BED_NAME,"+
                "ward_id                                                              AS ROOM_NO,"+
                "ward_name                                                            AS ROOM_NAME,"+
                "dept_code                                                            AS DEPT_ID,"+
                "dept_name                                                            AS DEPT_NAME,"+
                "wards_id                                                             AS WARD_ID,"+
                "wards_name                                                           AS WARD_NAME,"+
                "chief_complaint                                                      AS CHIEF_COMPLAINT,"+
                "current_history                                                      AS PRESENT_ILLNESS_HISTORY,"+
                "general_health_mark                                                  AS GENERAL_HEALTH_FLAG,"+
                "disease_history                                                      AS DISEASES_HISTORY,"+
                "infectious_mark                                                      AS INFECTIOUS_DISEASE_HISTORY,"+
                "infectious_disease_history                                           AS OBSTETRICAL_HISTORY,"+
                "marriage_history                                                     AS ALLERGIC_HISTORY,"+
                "allergy_history                                                      AS OPERATION_HISTORY,"+
                "surgery_history                                                      AS VACCINATION_HISTORY,"+
                "vaccination_history                                                  AS TRANSFUSION_HISTORY,"+
                "blood_history                                                        AS PERSONAL_HISTORY,"+
                "personal_history                                                     AS BIRTH_HISTORY,"+
                "menstrual_history                                                    AS MENSTRUAL_HISTORY,"+
                "family_history                                                       AS FAMILY_HISTORY,"+
                "temperature                                                          AS TEMPERATURE_VOLUME,"+
                "pulse_rate                                                           AS PULSE_RATE,"+
                "breathe                                                              AS RESPIRATORY_RATE,"+
                "systolic                                                             AS SYSTOLIC_PRESSURE,"+
                "diastolic                                                            AS DIASTOLIC_PRESSURE,"+
                "length                                                               AS PATIENT_HIGHT,"+
                "weight                                                               AS PATIENT_WEIGHT,"+
                "pe_all_exam                                                          AS LYMPHONODUS_EXAM_RESULT,"+
                "pe_ho_exam                                                           AS ORGANS_EXAM_RESULT,"+
                "pe_neck_exam                                                         AS NECK_EXAM_RESULT,"+
                "pe_chest_exam                                                        AS PE_CHEST_EXAM,"+
                "celiac_exam                                                          AS BELLY_EXAM_RESULT,"+
                "fingerprint_exam                                                     AS DIGTAL_EXAM_RESULT,"+
                "genitals_result                                                      AS GENITAL_CHECK_RESULT,"+
                "spinal_exam                                                          AS SPINE_EXAM_RESULT,"+
                "limb_exam                                                            AS ARMS_EXAM_RESULT,"+
                "pe_nervous_exame                                                     AS NERVOUS_SYSTEM_RESULT,"+
                "specialist_situation                                                 AS SPECIALIST_CASE,"+
                "item_result                                                          AS ASSIST_EXAM_TEST,"+
                "tcm_four_diagnosis                                                   AS CHINA_MED_OBSERVE_RESULT,"+
                "principle_and_method                                                 AS THERAPEUTIC_PRINCIPLE,"+
                "visit_date                                                           AS VISIT_DATE,"+
                "case when diag_type='初步诊断-西医诊断' then diag_date else '' end     AS PREL_DIAG_DATE,"+
                "case when diag_type='初步诊断-西医诊断' then diag_code else '' end     AS PREL_DIAG_ICD_CODE,"+
                "case when diag_type='初步诊断-西医诊断' then diag_name else '' end     AS PREL_DIAG_ICD_NAME,"+
                "case when diag_type='初步诊断-西医诊断' then diag_order else '' end    AS DIAG_ORDER,"+
                "case when diag_type='初步诊断-中医诊断' then diag_date else '' end     AS PREL_CHINA_DIAG_DATE,"+
                "case when diag_type='初步诊断-中医诊断' then diag_code else '' end     AS PREL_CHINA_DISEASE_CODE,"+
                "case when diag_type='初步诊断-中医诊断' then diag_name else '' end     AS PREL_CHINA_DISEASE_NAME,"+
                "case when diag_type='初步诊断-中医诊断' then diag_order else '' end    AS DIAG_ORDER_CZY,"+
                "case when diag_type='修正诊断-西医诊断' then diag_date else '' end     AS CORRECT_DIAG_DATE,"+
                "case when diag_type='修正诊断-西医诊断' then diag_code else '' end     AS CORRECT_DIAG_WESTERN_NAME,"+
                "case when diag_type='修正诊断-西医诊断' then diag_name else '' end     AS CORRECT_DIAG_WESTERN_CODE,"+
                "case when diag_type='修正诊断-西医诊断' then diag_order else '' end    AS DIAG_ORDER_XXY,"+
                "case when diag_type='修正诊断-中医诊断' then diag_date else '' end     AS CORRECT_CHINA_DIAG_DATE,"+
                "case when diag_type='修正诊断-中医诊断' then diag_code else '' end     AS CORRECT_DIAG_BMMEDICINE_CODE,"+
                "case when diag_type='修正诊断-中医诊断' then diag_name else '' end     AS CORRECT_DIAG_BMMEDICINE_NAME,"+
                "case when diag_type='修正诊断-中医诊断' then diag_order else '' end    AS DIAG_ORDER_XZY,"+
                "case when diag_type='补充诊断-西医诊断' then diag_date else '' end     AS SUPPLEMENT_DIAG_DATE,"+
                "case when diag_type='补充诊断-西医诊断' then diag_code else '' end     AS SUPPLEMENT_DIAG_WESTERN_NAME,"+
                "case when diag_type='补充诊断-西医诊断' then diag_name else '' end     AS SUPPLEMENT_DIAG_WESTERN_CODE,"+
                "case when diag_type='补充诊断-西医诊断' then diag_order else '' end    AS DIAG_ORDER_BXY," +
                "s.xds_type as XDS_TYPE," +
                "s.xds_name as XDS_NAME," +
                "s.xds_version as XDS_VERSION," +
                "visit_id AS VISIT_ID," +
                "'dwd_mhs_ipt_admissionnote' as table_name "+
                "from hdsd00_13_02 s " +
                "left join tdy_list_diag a on a.unique_id=s.unique_id and a.xds_type=s.xds_type");
        tableEnv.createTemporaryView("join_result11",joinResult11);

        Table joinResult12 = tableEnv.sqlQuery("SELECT a.unique_id    AS    RECORD_ID                                   "+
                "     ,a.hospital_name            AS     HOSPITAL_NAME                                                                   "+
                "     ,a.hospital_code            AS     HOSPITAL_CODE                                                                   "+
                "     ,a.age_unit                 AS     AGE_UNIT                                                                        "+
                "     ,a.age                      AS     AGE                                                                             "+
                "     ,a.sex                      AS     SEX_NAME                                                                        "+
                "     ,a.sex_code                 AS     SEX_CODE                                                                        "+
                "     ,a.identify_no              AS     ID_CARD_NUMBER                                                                  "+
                "     ,a.patient_typecode         AS     PATIENT_VISIT_CATEGORY_CODE                                                     "+
                "     ,a.patient_type             AS     PATIENT_VISIT_CATEGORY                                                          "+
                "     ,a.visit_id                 AS     INPATIENT_VISIT_FLOW_ID                                                         "+
                "     ,a.inpatient                AS     INPATIENT_NO                                                                    "+
                "     ,a.patient_id               AS     PATIENT_ID                                                                      "+
                "     ,a.patient_name             AS     PATIENT_NAME                                                                    "+
                "     ,a.domain_id                AS     DOMAIN_ID                                                                       "+
                "     ,a.visit_times              AS     VISIT_TIMES                                                                     "+
                "     ,a.birth                    AS     BIRTH_DATE                                                                      "+
                "     ,a.author_code              AS     AUTHOR_CODE                                                                     "+
                "     ,a.author_name              AS     AUTHOR_NAME                                                                     "+
                "     ,a.custodian_code           AS     CUSTODIAN_CODE                                                                  "+
                "     ,a.custodian_name           AS     CUSTODIAN_NAME                                                                  "+
                "     ,a.superior_signdate        AS     SENIOR_DOCTORE_SIGN_TIME                                                        "+
                "     ,a.superior_code            AS     SENIOR_DOCTOR_ID                                                                "+
                "     ,a.superior_name            AS     SENIOR_DOCTOR_SIGN_NAME                                                         "+
                "     ,a.resident_signdate        AS     RESIDENT_DOCTORE_SIGN_TIME                                                      "+
                "     ,a.resident_code            AS     RESIDENT_DOCTOR_ID                                                              "+
                "     ,a.resident_name            AS     RESIDENT_DOCTOR_SIGN_NAME                                                       "+
                "     ,a.visit_date               AS     VISIT_DATE                                                                      "+
                "     ,a.bed_no                   AS     BED_NO                                                                          "+
                "     ,a.bed_name                 AS     BED_NAME                                                                        "+
                "     ,a.ward_id                  AS     ROOM_NO                                                                         "+
                "     ,a.ward_name                AS     ROOM_NAME                                                                       "+
                "     ,a.wards_id                 AS     WARD_ID                                                                         "+
                "     ,a.wards_name               AS     WARD_NAME                                                                       "+
                "     ,a.dept_code                AS     DEPT_ID                                                                         "+
                "     ,a.dept_name                AS     DEPT_NAME                                                                       "+
                "     ,a.chief_complaint          AS     CHIEF_COMPLAINT                                                                 "+
                "     ,a.case_characteristics     AS     CASE_CHARACTER                                                                  "+
                "     ,a.tcm_four_diagnosis       AS     CHINA_MED_OBSERVE_RESULT                                                        "+
                "     ,a.diagnostic_basis         AS     DIAG_BASIS                                                                      "+
                "     ,a.assessment_plan          AS     TREAT_PLAN                                                                      "+
                "     ,a.principle_and_method     AS     THERAPEUTIC_PRINCIPLE                                                           "+
                "     ,case when b.diag_type='初步诊断-西医诊断' then b.diag_code else '' end    AS    PREL_DIAG_ICD_CODE                "+
                "     ,case when b.diag_type='初步诊断-西医诊断' then b.diag_name else '' end    AS    PREL_DIAG_ICD_NAME                "+
                "     ,case when b.diag_type='初步诊断-中医病名' then b.diag_code else '' end    AS    PREL_CHINA_DISEASE_CODE           "+
                "     ,case when b.diag_type='初步诊断-中医病名' then b.diag_name else '' end    AS    PREL_CHINA_DISEASE_NAME           "+
                "     ,case when b.diag_type='初步诊断-中医证候' then b.diag_code else '' end    AS    PREL_CHINA_SYNDROME_CODE          "+
                "     ,case when b.diag_type='初步诊断-中医证候' then b.diag_name else '' end    AS    PREL_CHINA_SYNDROME_NAME          "+
                "     ,case when b.diag_type='鉴别诊断-西医诊断' then b.diag_code else '' end    AS    DIFF_DIAG_ICD_CODE                "+
                "     ,case when b.diag_type='鉴别诊断-西医诊断' then b.diag_name else '' end    AS    DIFF_DIAG_ICD_NAME                "+
                "     ,case when b.diag_type='鉴别诊断-中医病名' then b.diag_code else '' end    AS    DIFF_CHINA_DISEASE_CODE           "+
                "     ,case when b.diag_type='鉴别诊断-中医病名' then b.diag_name else '' end    AS    DIFF_CHINA_DISEASE_NAME           "+
                "     ,case when b.diag_type='鉴别诊断-中医证候' then b.diag_code else '' end    AS    DIFF_CHINA_SYNDROME_CODE          "+
                "     ,case when b.diag_type='鉴别诊断-中医证候' then b.diag_name else '' end    AS    DIFF_CHINA_SYNDROME_NAME          " +
                "     ,a.xds_type as XDS_TYPE  " +
                "     ,a.xds_name as XDS_NAME  " +
                "     ,a.xds_version as XDS_VERSION  " +
                "     ,visit_id AS VISIT_ID " +
                "     ,'dwd_mhs_ipt_first_cod_record' as table_name "+
                " FROM hdsd00_14_14 a                                                                                                    "+
                " LEFT JOIN tdy_list_diag b                                                                                              "+
                " on a.unique_id = b.unique_id and a.xds_type = b.xds_type                                                             ");
        tableEnv.createTemporaryView("join_result12",joinResult12);
    }

    private void createSourceTables(StreamTableEnvironment tableEnv) {
       /* tableEnv.executeSql("create table hdsd00_08_05 (" +
                "pk                             String,"+
                "upload_time                    String,"+
                "status                         String,"+
                "empi                           String,"+
                "encounter_id                   String,"+
                "bed_name                       String,"+
                "ward_name                      String,"+
                "patient_id                     String,"+
                "patient_domain                 String,"+
                "visit_domain                   String,"+
                "visit_id                       String,"+
                "visit_times                    String,"+
                "unique_id                      String,"+
                "xds_type                       String,"+
                "xds_name                       String,"+
                "effective_time                 String,"+
                "xds_version                    String,"+
                "patient_type                   String,"+
                "patient_typecode               String,"+
                "out_patient_id                 String,"+
                "domain_id                      String,"+
                "inpatient                      String,"+
                "identify_no                    String,"+
                "patient_name                   String,"+
                "sex                            String,"+
                "sex_code                       String,"+
                "age                            String,"+
                "age_unit                       String,"+
                "institutional_incode           String,"+
                "institutional_in               String,"+
                "write_time                     String,"+
                "author_code                    String,"+
                "author_name                    String,"+
                "custodian_code                 String,"+
                "custodian_name                 String,"+
                "patrol_signdate                String,"+
                "patrol_code                    String,"+
                "patrol_name                    String,"+
                "instrument_signdate            String,"+
                "instrument_code                String,"+
                "instrument_name                String,"+
                "handover_code                  String,"+
                "handover_name                  String,"+
                "handover_signdate              String,"+
                "shipper_signdate               String,"+
                "shipper_code                   String,"+
                "shipper_name                   String,"+
                "visit_date                     String,"+
                "bed_no                         String,"+
                "ward_id                        String,"+
                "dept_code                      String,"+
                "dept_name                      String,"+
                "wards_id                       String,"+
                "wards_name                     String,"+
                "hospital_code                  String,"+
                "hospital_name                  String,"+
                "weight                         String,"+
                "abo_code                       String,"+
                "abo                            String,"+
                "rh_code                        String,"+
                "rh                             String,"+
                "skin_check                     String,"+
                "allergy_history_mark           String,"+
                "allergy_history                String,"+
                "nursing_gradecode              String,"+
                "nursing_grade                  String,"+
                "nursing_typecode               String,"+
                "nursing_type                   String,"+
                "patrol_date_ao                 String,"+
                "patrol_code_ao                 String,"+
                "patrol_name_ao                 String,"+
                "scrub_date_ao                  String,"+
                "scrub_code_ao                  String,"+
                "scrub_name_ao                  String,"+
                "oper_good_ao                   String,"+
                "preoperative_counting_mark     String,"+
                "patrol_date_ac                 String,"+
                "patrol_code_ac                 String,"+
                "patrol_name_ac                 String,"+
                "scrub_date_ac                  String,"+
                "scrub_code_ac                  String,"+
                "scrub_name_ac                  String,"+
                "oper_good_ac                   String,"+
                "before_check_mark              String,"+
                "patrol_date_bc                 String,"+
                "patrol_code_bc                 String,"+
                "patrol_name_bc                 String,"+
                "scrub_date_bc                  String,"+
                "scrub_code_bc                  String,"+
                "scrub_name_bc                  String,"+
                "oper_good_bc                   String,"+
                "after_check_mark               String"+
                ")" +
                FlinkSourceUtil.getKafkaWith("hdsd00_08_05","App_09_12"));*/
        tableEnv.executeSql("create table hdsd00_09_03 (" +
                "pk                    String,"+
                "upload_time           String,"+
                "status                String,"+
                "empi                  String,"+
                "encounter_id          String,"+
                "visit_date            String,"+
                "bed_name              String,"+
                "ward_name             String,"+
                "wards_id              String,"+
                "visit_domain          String,"+
                "visit_id              String,"+
                "visit_times           String,"+
                "patient_id            String,"+
                "patient_domain        String,"+
                "unique_id             String,"+
                "xds_type              String,"+
                "xds_name              String,"+
                "effective_time        String,"+
                "xds_version           String,"+
                "domain_id             String,"+
                "patient_type          String,"+
                "patient_typecode      String,"+
                "health_card           String,"+
                "inpatient             String,"+
                "identify_no           String,"+
                "patient_name          String,"+
                "sex_code              String,"+
                "sex                   String,"+
                "age_unit              String,"+
                "age                   String,"+
                "bed_no                String,"+
                "ward_id               String,"+
                "dept_code             String,"+
                "dept_name             String,"+
                "wards_name            String,"+
                "hospital_code         String,"+
                "hospital_name         String,"+
                "write_time            String,"+
                "author_code           String,"+
                "author_name           String,"+
                "custodian_code        String,"+
                "custodian_name        String,"+
                "doc_audit_date        String,"+
                "doc_auditorcode       String,"+
                "doc_auditor           String,"+
                "diag_code             String,"+
                "diag_name             String,"+
                "nursing_gradecode     String,"+
                "nursing_grade         String,"+
                "nursing_typecode      String,"+
                "nursing_type          String,"+
                "nursing_problem       String,"+
                "nurse_name            String,"+
                "nurse_category_item   String,"+
                "nurse_result          String,"+
                "catheter_description  String,"+
                "posture_nursing       String,"+
                "skin_care             String "+
                ")"+
                FlinkSourceUtil.getKafkaWith("hdsd00_09_03","App_09_12"));
        tableEnv.executeSql("create table hdsd00_13_02 (" +
                "pk                                         String,"+
                "upload_time                                String,"+
                "status                                     String,"+
                "empi                                       String,"+
                "encounter_id                               String,"+
                "patient_id                                 String,"+
                "patient_domain                             String,"+
                "visit_domain                               String,"+
                "visit_id                                   String,"+
                "visit_times                                String,"+
                "unique_id                                  String,"+
                "xds_type                                   String,"+
                "xds_name                                   String,"+
                "effective_time                             String,"+
                "xds_version                                String,"+
                "domain_id                                  String,"+
                "patient_type                               String,"+
                "patient_typecode                           String,"+
                "inpatient                                  String,"+
                "current_door_number                        String,"+
                "current_village                            String,"+
                "current_township                           String,"+
                "current_county                             String,"+
                "current_city                               String,"+
                "current_province                           String,"+
                "identify_no                                String,"+
                "patient_name                               String,"+
                "sex_code                                   String,"+
                "sex                                        String,"+
                "marital_code                               String,"+
                "marital_name                               String,"+
                "ethnic_code                                String,"+
                "ethnic                                     String,"+
                "age                                        String,"+
                "age_unit                                   String,"+
                "occupation_code                            String,"+
                "occupation                                 String,"+
                "write_time                                 String,"+
                "author_code                                String,"+
                "author_name                                String,"+
                "relationship_code                          String,"+
                "presenter_id_num                           String,"+
                "relationship_name                          String,"+
                "history_narrator                           String,"+
                "custodian_name                             String,"+
                "custodian_code                             String,"+
                "final_auditor_signdate                     String,"+
                "doc_final_auditorcode                      String,"+
                "doc_final_auditor                          String,"+
                "attend_signdate                            String,"+
                "admission_code                             String,"+
                "admission_name                             String,"+
                "resident_signdate                          String,"+
                "resident_code                              String,"+
                "resident_name                              String,"+
                "attending_signdate                         String,"+
                "attending_code                             String,"+
                "attending_name                             String,"+
                "admission_date                             String,"+
                "bed_no                                     String,"+
                "bed_name                                   String,"+
                "ward_id                                    String,"+
                "ward_name                                  String,"+
                "dept_code                                  String,"+
                "dept_name                                  String,"+
                "wards_id                                   String,"+
                "wards_name                                 String,"+
                "hospital_code                              String,"+
                "hospital_name                              String,"+
                "chief_complaint                            String,"+
                "current_history                            String,"+
                "general_health_mark                        String,"+
                "disease_history                            String,"+
                "infectious_mark                            String,"+
                "infectious_disease_history                 String,"+
                "marriage_history                           String,"+
                "allergy_history                            String,"+
                "surgery_history                            String,"+
                "vaccination_history                        String,"+
                "blood_history                              String,"+
                "personal_history                           String,"+
                "menstrual_history                          String,"+
                "family_history                             String,"+
                "temperature                                String,"+
                "pulse_rate                                 String,"+
                "breathe                                    String,"+
                "systolic                                   String,"+
                "diastolic                                  String,"+
                "length                                     String,"+
                "weight                                     String,"+
                "general_exam                               String,"+
                "skin_mucosal_exam                          String,"+
                "pe_all_exam                                String,"+
                "pe_ho_exam                                 String,"+
                "pe_neck_exam                               String,"+
                "pe_chest_exam                              String,"+
                "celiac_exam                                String,"+
                "fingerprint_exam                           String,"+
                "genitals_result                            String,"+
                "spinal_exam                                String,"+
                "limb_exam                                  String,"+
                "pe_nervous_exame                           String,"+
                "specialist_situation                       String,"+
                "item_result                                String,"+
                "statement_reliability_mark                 String,"+
                "tcm_four_diagnosis                         String,"+
                "principle_and_method                       String,"+
                "visit_date                                 String "+
                ")"+
                FlinkSourceUtil.getKafkaWith("hdsd00_13_02","App_09_12"));
        tableEnv.executeSql("create table hdsd00_14_14 (       " +
                "pk                       string," +
                "upload_time              string," +
                "status                   string," +
                "empi                     string," +
                "encounter_id             string," +
                "patient_id               string," +
                "patient_domain           string," +
                "visit_domain             string," +
                "visit_id                 string," +
                "visit_times              string," +
                "unique_id                string," +
                "xds_type                 string," +
                "xds_name                 string," +
                "effective_time           string," +
                "xds_version              string," +
                "domain_id                string," +
                "patient_type             string," +
                "patient_typecode         string," +
                "inpatient                string," +
                "identify_no              string," +
                "patient_name             string," +
                "sex_code                 string," +
                "sex                      string," +
                "birth                    string," +
                "age                      string," +
                "age_unit                 string," +
                "write_time               string," +
                "author_code              string," +
                "author_name              string," +
                "custodian_code           string," +
                "custodian_name           string," +
                "superior_signdate        string," +
                "superior_code            string," +
                "superior_name            string," +
                "resident_signdate        string," +
                "resident_code            string," +
                "resident_name            string," +
                "visit_date               string," +
                "bed_no                   string," +
                "bed_name                 string," +
                "ward_id                  string," +
                "ward_name                string," +
                "wards_id                 string," +
                "wards_name               string," +
                "dept_code                string," +
                "dept_name                string," +
                "hospital_code            string," +
                "hospital_name            string," +
                "chief_complaint          string," +
                "case_characteristics     string," +
                "tcm_four_diagnosis       string," +
                "diagnostic_basis         string," +
                "assessment_plan          string," +
                "principle_and_method     string" +
                ")" +
                FlinkSourceUtil.getKafkaWith("hdsd00_14_14", "App_09_12"));


        super.readTdyListDiag(tableEnv,"App_09_12");
        super.readTdyListNurseTrachea(tableEnv,"App_09_12");
        super.readTdyListNurseSafety(tableEnv,"App_09_12");
        super.readTdyListDietGuide(tableEnv,"App_09_12");
//        super.readTdyListHandover(tableEnv,"App_09_12");
        super.readTdyListOper(tableEnv,"App_09_12");
    }
}
